package com.evamay.kotlindemo.ui.rxdemo;

import android.os.Bundle;

import com.evamay.kotlindemo.R;
import com.evamay.kotlindemo.entity.Developer;
import com.evamay.kotlindemo.entity.Human;
import com.evamay.kotlindemo.entity.TypeInfo;
import com.evamay.kotlindemo.ui.main.MainAdapter;
import com.orhanobut.logger.Logger;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;

import androidx.annotation.Nullable;
import androidx.appcompat.app.AppCompatActivity;
import androidx.recyclerview.widget.LinearLayoutManager;
import androidx.recyclerview.widget.RecyclerView;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class RxDemoByJavaActivity extends AppCompatActivity {

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_demo);
        //创建Observer观察者，它决定事件触发时有怎样的行为
        Observer observer=new Observer() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };
        //Subscriber事件订阅接口，对Observer进行了一些扩展，与Observer用法一致
        Subscriber subscriber=new Subscriber() {
            @Override
            public void onSubscribe(Subscription s) {

            }

            @Override
            public void onNext(Object o) {

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }

        };
        initData();



    }

    private void initData() {
        RecyclerView rv=findViewById(R.id.rv_rx_by_java);
        final List<TypeInfo> list=new ArrayList<>();
        list.add(new TypeInfo(1,"入门使用"));
        list.add(new TypeInfo(2,"filter过滤操作符"));
        list.add(new TypeInfo(3,"map数据类型转换操作符"));
        list.add(new TypeInfo(4,"线程调度"));
        LinearLayoutManager layoutManager=new LinearLayoutManager(this);
        MainAdapter mAdapter=new MainAdapter(this,list);
        rv.setLayoutManager(layoutManager);
        rv.setAdapter(mAdapter);
        mAdapter.setClickListener(new MainAdapter.OnItemClickListener() {
            @Override
            public void onItemClick(int position) {
                switch (list.get(position).getType()) {
                    case 1:
                        helloRx();
                        break;
                    case 2:
                        filterOperating();
                        break;
                    case 3:
                        mapOperating();
                        break;
                    case 4:
                        threadSchedulers();
                        break;
                }
            }
        });
    }

    /**
     * 线程调度
     */
    private void threadSchedulers() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Logger.e("subscribe 的当前线程: "+Thread.currentThread().getName());
                emitter.onNext("这是一个线程调度操作");
                emitter.onComplete();
            }
        })
                .subscribeOn(Schedulers.io())//指定subscribe方法在IO线程池中运行
                .observeOn(Schedulers.newThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Logger.e("onSubscribe: 当前线程："+Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(String s) {
                        Logger.e("onNext: 当前线程："+Thread.currentThread().getName()+" s: "+s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Logger.e("onError: 当前线程："+Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        Logger.e("onComplete: 当前线程："+Thread.currentThread().getName());
                    }
                });
    }



    /**
     * map数据类型转换操作符
     */
    private void mapOperating() {
        Observer<Developer> developerObserver=new Observer<Developer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Logger.e("Disposable ");
            }

            @Override
            public void onNext(Developer developer) {
                Logger.e("onNext: developer: "+developer.toString());
            }

            @Override
            public void onError(Throwable e) {
                Logger.e("onError: "+e.getMessage());
            }

            @Override
            public void onComplete() {
                Logger.e("onComplete");
            }
        };
        Human human=new Human();
        human.setName("李狗嗨");
        human.setAge(35);
        Observable.just(human).map(new Function<Human, Developer>() {
            @Override
            public Developer apply(Human human) throws Exception {
                Developer developer=new Developer();
                developer.setName(human.getName());
                developer.setAge(human.getAge());
                developer.setSkill("李狗蛋");
                return developer;
            }
        }).subscribe(developerObserver);
    }

    /**
     * filter过滤操作符
     */
    private void filterOperating() {
        Observer<String> observer=new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Logger.e("onSubscribe: ");
            }

            @Override
            public void onNext(String s) {
                //这里只允许满足过滤条件的通过
                Logger.e("onNext: "+s);
            }

            @Override
            public void onError(Throwable e) {
                Logger.e("onError: "+e.getMessage());
            }

            @Override
            public void onComplete() {
                Logger.e("onComplete: ");
            }

        };
        Observable.just("炒饭","炒面","炒河粉","凉皮","凉面","酸辣粉")
                .filter(new Predicate<String>() {
                    @Override
                    public boolean test(String s) throws Exception {
                        return s.equals("酸辣粉");
                    }
                }).subscribe(observer);

    }

    /**
     * 最基本入门
     */
    private void helloRx() {
        //最基础运用
        Consumer<String> consumer=new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Logger.e(s);
            }
        };
        Observable.just("hello").subscribe(consumer);

        //进阶运用
        //创建观察者
        Observer<String> mObserver=new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Logger.e("onSubscribe: ");
            }

            @Override
            public void onNext(String s) {
                Logger.e("onNext: "+s);
            }

            @Override
            public void onError(Throwable e) {
                Logger.e("onError: "+e.getMessage());
            }

            @Override
            public void onComplete() {
                Logger.e("onComplete ");
            }
        };
        //被观察者发出消息，通过subscribe订阅来指定该事件的观察者mObserver
        Observable.just("准备发射").subscribe(mObserver);

        //进阶升级
        Observable<String> observable=new Observable<String>() {
            @Override
            protected void subscribeActual(Observer<? super String> observer) {
                observer.onNext("10");
                observer.onNext("9");
                observer.onNext("8");
                observer.onNext("7");
                observer.onNext("6");
                observer.onNext("5");
                observer.onNext("4");
                observer.onNext("3");
                observer.onNext("2");
                observer.onNext("1");
                observer.onNext("点火");
                observer.onNext("起飞");
            }
        };

        observable.subscribe(mObserver);
    }


}
